package com.ccb.textinputformat;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import com.ccb.posrecord.PosCustomRecordReader;
import com.google.common.base.Charsets;

public class PosCustomTextInputFormat extends TextInputFormat {
	@Override
	public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) {
		String delimiter = context.getConfiguration().get("textinputformat.record.delimiter");
		byte[] recordDelimiterBytes = null;
		if (null != delimiter)
			recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
		try {
			return new PosCustomRecordReader(Class.forName(context.getConfiguration().get("customRecord")), recordDelimiterBytes);
		} catch (ClassNotFoundException e) {
			throw new RuntimeException("ClassNotFoundException:" + context.getConfiguration().get("customRecord"));
		}
	}
}
